package com.ssl.canal.listener;

import com.alibaba.fastjson.JSONObject;
import com.ssl.canal.pojo.CanalMessage;
import com.ssl.canal.utils.RedisCommonProcessor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
public class CanalRabbitMqListener {
    @Autowired
    private RedisCommonProcessor redisCommonProcessor;

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue(value = "canal.queue",durable = "true"),
                    exchange = @Exchange(value = "canal.exchange"),
                    key = "canal.routing.key"
            )
    })
    public void handleDataChange(String message){
        try{
            CanalMessage msg = JSONObject.parseObject(message, CanalMessage.class);
            if (msg.getTable().equals("user")&& msg.getDatabase().equals("oauth")
                    && !msg.getType().equals("INSERT")) {
                List<Map<String, Object>> data = msg.getData();
                for (Map m : data){
                    Integer id = Integer.valueOf(m.get("id").toString());
                    if (id != null){
                        redisCommonProcessor.del(id + 10000000+"");
                    }
                }
            }
        } catch (Exception e){
            //todo
        }

    }
}
